package com.ysb.hbase.config;

import com.ysb.hbase.dml.HbaseDmlService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.security.UserGroupInformation;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
@Slf4j
@EnableConfigurationProperties(HbaseProperties.class)
public class HbaseAutoConfiguration implements ApplicationContextAware {

    private static final String HBASE_QUORUM = "hbase.zookeeper.quorum";
    private static final String HBASE_CLIENT_PORT = "hbase.zookeeper.property.clientPort";
    private static final String HBASE_ROOT_DIR = "hbase.rootdir";
    private static final String HBASE_ZNODE_PARENT = "zookeeper.znode.parent";
    private static final String HBASE_CLIENT_RETRIES_NUMBER = "hbase.client.retries.number";

    private static final String HBASE_RPC_TIMEOUT = "hbase.rpc.timeout";
    private static final String HBASE_CLIENT_PAUSE = "hbase.client.pause";
    private static final String HBASE_CLIENT_OPERATION_TIMEOUT = "hbase.client.operation.timeout";
    private ThreadPoolExecutor executor;

    @Autowired
    private HbaseProperties hbaseProperties;

    @Value("${spring.profiles.active}")
    private String env;
    private ApplicationContext applicationContext;

    @Bean
    @ConditionalOnMissingBean(HbaseDmlService.class)
    public HbaseDmlService hbaseDmlService() {
        String krb5ConfLocal = hbaseProperties.getKrb5ConfLocal();
        if (StringUtils.isBlank(krb5ConfLocal)) {
            String confName = "";
            if ("prod".equals(env)) {
                confName = "krb5_pro.conf";
            } else {
                confName = "krb5_dev.conf";
            }
            URL confUrl = this.getClass().getClassLoader().getResource(confName);
            if (confUrl == null) {
                throw new RuntimeException("hbaseDmlService getResource: krb5_pro.conf not exists");
            }
            krb5ConfLocal = confUrl.getPath();
        } else {
            File file = new File(krb5ConfLocal);
            if (!file.isFile()) {
                throw new RuntimeException("hbaseDmlService krb5ConfLocal：" + krb5ConfLocal + " not exists");
            }
        }
        System.setProperty("java.security.krb5.conf", krb5ConfLocal);

        org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
        configuration.set(HBASE_QUORUM, hbaseProperties.getQuorum());
        configuration.set(HBASE_CLIENT_PORT, hbaseProperties.getClientPort());
        configuration.set(HBASE_ROOT_DIR, "/" + hbaseProperties.getRootDir());
        configuration.set(HBASE_ZNODE_PARENT, "/" + hbaseProperties.getNodeParent());

        Integer retriesNum = hbaseProperties.getRetriesNum();
        configuration.setInt(HBASE_CLIENT_RETRIES_NUMBER, retriesNum == null || retriesNum == 0 ? 6 : retriesNum);
        Integer rpcTimeout = hbaseProperties.getRpcTimeout();
        configuration.setInt(HBASE_RPC_TIMEOUT, rpcTimeout == null || rpcTimeout == 0 ? 100 : rpcTimeout);
        Integer clientPause = hbaseProperties.getClientPause();
        configuration.setInt(HBASE_CLIENT_PAUSE, clientPause == null || clientPause == 0 ? 100 : clientPause);
        Integer operationTimeout = hbaseProperties.getOperationTimeout();
        configuration.setInt(HBASE_CLIENT_OPERATION_TIMEOUT, operationTimeout == null || operationTimeout == 0 ? 100 : operationTimeout);

        configuration.set("hbase.security.authentication", "kerberos");
        configuration.set("hadoop.security.authentication", "kerberos");

        configuration.set("hbase.master.kerberos.principal", "root/_HOST@JENKIN.COM");
        configuration.set("hbase.regionserver.kerberos.principal", "root/_HOST@JENKIN.COM");

        UserGroupInformation.setConfiguration(configuration);
        try {
            String keyTab = hbaseProperties.getKeyTabLocal();
            if (StringUtils.isBlank(keyTab)) {
                String keytabName = "";
                if ("prod".equals(env)) {
                    keytabName = "app_pro.keytab";
                } else {
                    keytabName = "app_dev.keytab";
                }
                URL keyTabUrl = this.getClass().getClassLoader().getResource(keytabName);
                if (keyTabUrl == null) {
                    throw new RuntimeException("hbaseDmlService getResource：app_pro.keytab not exists");
                }
                keyTab = keyTabUrl.getPath();
            } else {
                File file = new File(keyTab);
                if (!file.isFile()) {
                    throw new RuntimeException("hbaseDmlService keyTab：" + keyTab + " not exists");
                }
            }

            String principal = hbaseProperties.getPrincipal();
            if (StringUtils.isBlank(principal)) {
                principal = "app/ysbang@JENKIN.COM";
            }

            UserGroupInformation.loginUserFromKeytab(principal, keyTab);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        try {
            ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(50, 200, 60L,
                    TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
            // init pool
            poolExecutor.prestartCoreThread();
            poolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            Connection connection = ConnectionFactory.createConnection(configuration, poolExecutor);
            executor = poolExecutor;
            return new HbaseDmlService(configuration, connection);
        } catch (Exception e) {
            log.error("hbase connection资源池创建失败", e);
        }

        return new HbaseDmlService(configuration, null);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

